
#include <iostream>

#include <boost/thread/thread.hpp>
#include <boost/bind.hpp>

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/thread/mutex.hpp>

#include "bocat/common/base64.hpp"

#include "boque/buffesq.hpp"
#include "boque/trasit/broker.hpp"
#include "boque/trasit/channel.hpp"
#include "boque/trasit/client.hpp"

namespace boque		{
namespace trasit	{


///////////////////////////////////////////////////////////////////////
//

class	Tube
{

public	:
	
	Tube( const std::string& address, const std::string buf_dir )
		:	_address( address ),
			_buf_dir( buf_dir ),
			//_buffesq( buf_dir ),
			_buffesq_p( NULL ),
			_broker_p( NULL )
	{
		std::string protocol, host, port;

		if ( _buf_dir.size() > 1 ) {
			_buffesq_p = new BuffesQ( _buf_dir );
		}

		// address format : tcp://127.0.0.1:5555
		Channel::parseAddress( address, protocol, host, port );

		_broker_p = new Broker( host, port );
		_broker_connected = false;
		_broker_retry_mark = 0;
	}

	~Tube()
	{
		if ( NULL != _broker_p ) {
			_broker_p->disconnect();
			delete _broker_p;
			_broker_p = NULL;
		}

		if ( NULL != _buffesq_p ) {
			delete _buffesq_p;
			_buffesq_p = NULL;
		}
	}

public	:

	int		send( const std::string& data, bool bufferable=true )
	{
		return	send( data.c_str(), data.size(), bufferable );
	}

	int		send( const char* data, size_t size, bool bufferable=true )
	{
		boost::mutex::scoped_lock   _scoped_lock( _mutex );

		std::string request = bocat::common::base64_encode( data, size );

		if ( (true == bufferable) && (NULL == _buffesq_p) ) {
			bufferable = false;
		}

		if ( bufferable ) {
			if ( false == _broker_connected ) {
				++_broker_retry_mark;
				if ( canTryReconnect() ) {
					try {
						_broker_p->connect();

						_broker_connected = true;
						_broker_retry_mark = 0;
					}
					catch( std::exception& e ) {
						_broker_connected = false;
					}
				}
			}

			if ( false == _broker_connected ) {
				// push data to disk queue
				//_buffesq.push( request );
				_buffesq_p->push( request );

				return	0;
			}
		}

		bool	done = true;
        try {
            int rc = _broker_p->send( request );
            if ( rc < 0 ) { 
                //bocat::common::Logger*  logger_p = bocat::common::getLogger();
                //logger_p->error(__LINE__, __FILE__, "fail to send data to %s ", _address.c_str());
				
				done = false;
            }   
        }   
        catch( std::exception& e ) { 
            //bocat::common::Logger*  logger_p = bocat::common::getLogger();
            //logger_p->error(__LINE__, __FILE__, "fail to send data to %s -- %s", _address.c_str(), e.what());
            //logger_p->error(__LINE__, __FILE__, "channel to %s enter BUFFER mode", _address.c_str());

            _broker_connected = false;

			done = false;
        }   

		if ( !done ) {
			if ( !bufferable ) {
				return	-1;
			}

            // push data to disk queue
            //_buffesq.push( request );
            _buffesq_p->push( request );
		}

		return	0;
	}

private	:
	
	bool	canTryReconnect( void )
	{
		if ( _broker_retry_mark < 10 ) {
			return	true;
		}
		else if ( _broker_retry_mark < 50 ) {
			if ( 0 == (_broker_retry_mark % 5) ) {
				return	true;
			}
		}
		else if ( _broker_retry_mark < 200 ) {
			if ( 0 == (_broker_retry_mark % 10) ) {
				return	true;
			}
		}
		else {
			if ( 0 == (_broker_retry_mark % 20) ) {
				return	true;
			}
		}

		return	false;
	}

private	:
	std::string     _address;
	std::string     _buf_dir;
	//BuffesQ         _buffesq;
	BuffesQ*        _buffesq_p;
	Broker*         _broker_p;
	boost::mutex    _mutex;
	
	bool	_broker_connected;
	uint32_t	_broker_retry_mark;
};	//// class Tube


///////////////////////////////////////////////////////////////////////
//

Client::Client( const std::string& buf_dir )
	:	_buf_dir( buf_dir )
{
}

Client::~Client()
{
	tube_iterator_type	it  = _tubes.begin();
	tube_iterator_type	end = _tubes.end();
	for ( ; it != end; ++it ) {
		void*	tube_vp = it->second;

		Tube*	tube_p = static_cast< Tube* >( tube_vp );
		
		delete tube_p;
	}

	_tubes.clear();
}


///////////////////////////////////////////////////////////////////////
//

int
Client::send( const std::string& to, const std::string& data, bool bufferable )
{
	tube_v_ptr	tube_vp = getTube( to );
	Tube*	tube_p = static_cast< Tube* >( tube_vp );

	int	rc = tube_p->send( data, bufferable );

	return	rc;
}

///////////////////////////////////////////////////////////////////////
//


Client::tube_v_ptr
Client::getTube( const std::string& addr )
{
	tube_iterator_type	result = _tubes.find( addr );
	if ( _tubes.end() != result ) {
		// found
		return	result->second;
	}

	std::string	tube_dir;

	if ( _buf_dir.size() > 1 ) {
		std::string	dir_s = addr;
		boost::replace_all( dir_s, ":", "%" );
		boost::replace_all( dir_s, "/", "%" );

		tube_dir = _buf_dir + "/" + dir_s;

		boost::filesystem::path	p( tube_dir );
		if ( !boost::filesystem::exists( p ) ) {
			try {
				boost::filesystem::create_directories( p );
			}   
			catch ( std::exception& e ) { 
				std::ostringstream  msg;
				msg << "fail to create tube buf directory " << tube_dir << " -- " << e.what();
				throw   std::runtime_error( msg.str() );
			}   
		}
	}

	Tube*	tube_p = new Tube( addr, tube_dir );

	_tubes.insert( std::make_pair( addr, tube_p ) );

	return	tube_p;
}


///////////////////////////////////////////////////////////////////////
//

}	//// namespace trasit
}	//// namespace boque

